package com.navinfo.platform.bigdata.hive.udtf;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.navinfo.platform.bigdata.hive.dto.VehicleStatusAdditionDto;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

import static com.navinfo.platform.bigdata.hive.dto.LCStatusType.StatusType.*;

/**
 * 一对多类型：接受零个或多个输入，然后产生多列或多行的输出
 * UDTF：用户自定义表生成函数，如explode()，需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF；
 */
@Description(
	name = "status_addition",
	value = "_FUNC_(col) - The parameter is a column name. The return value is two strings.",
	extended = "Example:\n"
		+ " > SELECT _FUNC_(col) FROM src;"
		+ " > SELECT _FUNC_(col) AS (name, surname) FROM src;"
		+ " > SELECT adTable.name,adTable.surname FROM src LATERAL VIEW _FUNC_(col) adTable AS name, surname;"
)
public class AlarmAdditionUDF extends GenericUDTF{
    private String[] result;
	/**
     * 该方法指定输入输出参数：输入的Object Inspectors和输出的Struct。
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
	@Override
	public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
		if(argOIs.length != 1){
			throw new UDFArgumentException("StatusAdditionUDF takes exactly one argument.");
		}

		if(argOIs[0].getCategory() != ObjectInspector.Category.PRIMITIVE
				&& ((PrimitiveObjectInspector)argOIs[0]).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){
			throw new UDFArgumentTypeException(0, "StatusAdditionUDF takes a string as a parameter.");
		}

		ArrayList<String> fieldNames = new ArrayList<>();
		fieldNames.add("handBrakeStatus");
		fieldNames.add("slideNeutralStatus");
		fieldNames.add("rapidAccelerationStatus");
		fieldNames.add("rapidDecelerationStatus");
		fieldNames.add("sharpTurningStatus");
		fieldNames.add("slamAcceleratorStatus");


		ArrayList<ObjectInspector> fieldOIs = new ArrayList<>();
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
	}

    /**
     * 该方法处理输入记录，然后通过forward()方法返回输出结果
     * @param args
     * @throws HiveException
     */
	@Override
	public void process(Object[] args) throws HiveException {
		result = new String[]{
				"0","0","0","0","0",
				"0"};
		String input = args[0].toString();

		if (StringUtils.startsWith(input, "[")) {
			List<Byte> dataList = JSON.parseObject(input,  new TypeReference<List<Byte>>() {});
			if (dataList.size() >= 8) {
				//在第8个字节 Bit1 手刹开关
				result[0] = String.valueOf(((dataList.get(7) >> 1) & 0x01));
			}
			if (dataList.size() >= 6) {
				//在第6个字节
				byte additionAlarmAt6 = dataList.get(5);
				//bit0 空挡滑行
				result[1] = String.valueOf( additionAlarmAt6 & 0x01 );
				//bit1 急加速
				result[2] = String.valueOf( (additionAlarmAt6 >> 1) & 0x01 );
				//bit2 急减速
				result[3] = String.valueOf( (additionAlarmAt6 >> 2) & 0x01 );
				//bit3 急转弯
				result[4] = String.valueOf( (additionAlarmAt6 >> 3) & 0x01 );
				//bit6 猛踩油门报警状态
				result[5] = String.valueOf( (additionAlarmAt6 >> 6) & 0x01 );
			}
		}


		forward(result);
	}

    /**
     * 用于通知UDTF没有行可以处理了。可以在该方法中清理代码或者附加其他处理输出。
     * @throws HiveException
     */
	@Override
	public void close() throws HiveException {
		result = null;
	}

}
